package com.linys.scala.qf.day06_spark

import org.apache.spark.{SparkConf, SparkContext}

/**
  * 聚合进阶---理解理解
  */
object ExeciseAggregate_03 {

  def func1(index: Int, iter: Iterator[(Int)]) : Iterator[String] = {
    iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
  }

  def func2(index: Int, iter: Iterator[(String)]) : Iterator[String] = {
    iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
  }

  def func3(index: Int, iter: Iterator[(String, Int)]) : Iterator[String] = {
    iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
  }

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("ExeciseAggregate").setMaster("local[*]")
    val sc = new SparkContext(conf)

    println("--------------------------------")
    val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
    println(rdd1.mapPartitionsWithIndex(func1).collect.toBuffer)
    println(rdd1.aggregate(0)(math.max(_, _), _ + _))
    println(rdd1.aggregate(5)(math.max(_, _), _ + _))

    println("--------------------------------")
    val rdd2 = sc.parallelize(List("a","b","c","d","e","f"),2)
    println(rdd2.mapPartitionsWithIndex(func2).collect.toBuffer)
    println(rdd2.aggregate("")(_ + _, _ + _))
    println(rdd2.aggregate("=")(_ + _, _ + _))

    println("--------------------------------")
    val rdd3 = sc.parallelize(List("12","23","345","4567"),2)
    println(rdd3.mapPartitionsWithIndex(func2).collect.toBuffer)
    println(rdd3.aggregate("")((x,y) => math.max(x.length, y.length).toString, (x,y) => x + y))

    println("--------------------------------")
    val rdd4 = sc.parallelize(List("12","23","345",""),2)
    println(rdd4.mapPartitionsWithIndex(func2).collect.toBuffer)
    println(rdd4.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y))

    println("--------------------------------")
    val rdd5 = sc.parallelize(List("12","23","","345"),2)
    println(rdd5.mapPartitionsWithIndex(func2).collect.toBuffer)
    println(rdd5.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y))


    println("--------------------------------")
    val pairRDD = sc.parallelize(List(("mouse", 2),("cat",2), ("cat", 5), ("mouse", 4),("cat", 12), ("dog", 12)), 2)
    println(pairRDD.mapPartitionsWithIndex(func3).collect.toBuffer)
    println(pairRDD.aggregateByKey(0)(math.max(_, _), _ + _).collect.toBuffer)
    println(pairRDD.aggregateByKey(100)(math.max(_, _), _ + _).collect.toBuffer)

    sc.stop()

  }

}
